package raft

import "time"

// AppendEntriesArgs 是 AppendEntries RPC 的参数
type AppendEntriesArgs struct {
	Term         int        // 当前节点的 Term
	LeaderId     int        // 使 follower 可以为客户端重定向
	PrevLogIndex int        // 紧接在新日志之前的日志条目的索引
	PrevLogTerm  int        // 紧邻新日志条目之前的日志条目的任期
	LeaderCommit int        // leader 的已知已提交的最高的日志条目的索引
	Entries      []LogEntry // 需要被保存的日志条目（做心跳使用时，内容为空；为了提高效率可一次性发送多个）
}

// AppendEntriesReply 是 AppendEntries RPC 的返回值
type AppendEntriesReply struct {
	Success       bool // 是否添加成功
	Term          int  // 当前节点的 Term
	ConflictTerm  int  // 发生冲突的 LogEntry 的 Term
	ConflictIndex int  // 发生冲突的 LogEntry 的索引
}

// leaderSendAppendEntries  用于向指定的服务器发送 AppendEntries RPC
func (rf *Raft) leaderSendAppendEntries(server int) {
	rf.mu.RLock()
	// 如果这个请求的节点不是leader直接返回
	if rf.state != Leader {
		rf.mu.RUnlock()
		return
	}

	// 获取下一个需要发送的日志条目索引
	startIndex := rf.nextIndex[server]
	// 如果要发送的节点的Index比log的第一条的index还小，直接返回
	// 原因是后面如果做了 Snapshot，这部分数据不在内存中，等着直接拷贝Snapshot过去就好了，
	// 正常情况下不会出现这种问题，没有做 Snapshot 的话，所有 logEntries 都在内存
	if startIndex <= rf.log[0].Index {
		rf.mu.RUnlock()
		return
	}

	// 获取 需要作为参数的
	// 紧接在新日志之前的日志条目的索引
	prevLogIndex := startIndex - 1
	// 紧邻新日志条目之前的日志条目的任期
	prevLogTerm := rf.GetLogEntry(prevLogIndex).Term

	// 获取需要发送的日志条目
	entries := make([]LogEntry, len(rf.log[rf.GetIndex(startIndex):]))
	copy(entries, rf.log[rf.GetIndex(startIndex):])

	// 构造 AppendEntriesArgs
	appendEntriesArgs := AppendEntriesArgs{
		Term:         rf.currentTerm,
		LeaderId:     rf.me,
		LeaderCommit: rf.commitIndex,
		Entries:      entries,
		PrevLogIndex: prevLogIndex,
		PrevLogTerm:  prevLogTerm,
	}
	appendEntriesReply := AppendEntriesReply{}

	rf.mu.RUnlock()

	// 发送 AppendEntriesRPC请求
	ok := rf.sendAppendEntries(server, &appendEntriesArgs, &appendEntriesReply)
	if !ok {
		return
	}

	rf.mu.Lock()

	// 处理返回结果
	success := appendEntriesReply.Success
	term := appendEntriesReply.Term
	conflictTerm := appendEntriesReply.ConflictTerm
	conflictIndex := appendEntriesReply.ConflictIndex

	// 返回节点的 term 大于当前节点的 term，当前节点过期，自动转为 follower
	if term > rf.currentTerm {
		rf.state = Follower
		rf.votedFor = -1
		rf.currentTerm = term
		rf.persist()
	}

	// 如果当前节点不是 leader 或者 term != currentTerm，说明当前节点已经不是 leader了，直接返回
	if rf.state != Leader || term != rf.currentTerm {
		rf.mu.Unlock()
		return
	}

	if success { // AppendEntries 添加 LogEntries 成功的情况
		rf.matchIndex[server] = prevLogIndex + len(entries) // 更新 matchIndex
		rf.nextIndex[server] = rf.matchIndex[server] + 1    // 更新 nextIndex

		// DPrintf("leaderSendAppendEntries(): [%d] appendEntries to server[%d] success", rf.me, server)

		// 统计暂未被统计提交的 LogEntries 在各个节点上被添加的个数，超过半数则通知应用
		for i := rf.GetLogEntry(-1).Index; i > rf.commitIndex; i-- {
			if rf.GetLogEntry(i).Term != rf.currentTerm {
				continue
			}
			// 统计某一 未被提交的 LogEntries 被多少节点已经添加，超过半数可以应用
			matchCount := 0
			for _, matchIndex := range rf.matchIndex {
				if matchIndex >= i {
					matchCount += 1
				}
			}
			// 如果超过半数
			if matchCount > len(rf.peers)/2 {
				rf.commitIndex = i // 更新提交的索引
				rf.applyCond.Signal()
				break // break 可以省掉很多统计，因为如果后面的 index 已经超过半数并且被添加，前面的 index 也肯定被添加了
			}
		}
	} else if conflictIndex != 0 { // 如果出现LogEntry冲突，即 Term 和 Index 对应不一样
		rf.nextIndex[server] = conflictIndex // 将 nextIndex 重置为 发生冲突的那个 Index
		if conflictTerm > 0 {                // 如果冲突的 Term 也有问题，不是在当前 Term 出问题
			for i := rf.GetLogEntry(-1).Index; i > rf.log[0].Index; i -= 1 {
				// 找到 最后一个为当前冲突 Term 的 index
				if rf.GetLogEntry(i).Term == conflictTerm {
					rf.nextIndex[server] = i + 1
					break
				}
			}
		}
	}
	rf.mu.Unlock()
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) error {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	defer rf.persist()

	// 如果 Leader 的 Term 比当前节点的 Term 小，说明 leader 过期了
	leaderTerm := args.Term
	if leaderTerm < rf.currentTerm {
		reply.Success = false
		reply.Term = rf.currentTerm
		return nil
	}

	// 如果 Leader 的 Term 比当前节点的 Term 大，更新当前节点的 Term，并转换为 Follower
	if leaderTerm > rf.currentTerm {
		rf.currentTerm = leaderTerm
		rf.votedFor = -1
	}
	rf.state = Follower
	rf.lastHeartbeat = time.Now()

	// 做到这里其实 Lab2A 就已经可以 PASS了，只需要更新心跳就好了
	// Lab2A 并没有要求实现 log 的复制细节，已经冲突等，这是 Lab2B的内容

	// 获取 Leader 的 发送的 Args
	leaderLogEntries := args.Entries
	prevLogIndex := args.PrevLogIndex
	prevLogTerm := args.PrevLogTerm

	//  prevLogIndex 比当前节点的 最小 LogEntries 还小，
	// Leader 发送的 已过期，直接返回
	if prevLogIndex < rf.log[0].Index {
		reply.Success = false
		reply.Term = rf.currentTerm
		return nil
	}

	// prevLogIndex 比当前节点的最后一个条 Log 还大，
	// Leader 发送的 LogEntry 与当前节点冲突，需要重前面开始校对
	if prevLogIndex > rf.GetLogEntry(-1).Index {
		reply.Success = false
		reply.Term = rf.currentTerm
		reply.ConflictTerm = -1
		reply.ConflictIndex = rf.GetLogEntry(-1).Index + 1
		return nil
	}

	// prevLogIndex 对应的 Term 和 当前节点中对应的索引位置的 Term 不一致，
	// Leader 发送的 LogEntry 相同索引位置的日志不一致，冲突，需要重新校对
	if rf.GetLogEntry(prevLogIndex).Term != prevLogTerm {
		reply.Success = false
		reply.Term = rf.currentTerm
		reply.ConflictTerm = rf.GetLogEntry(prevLogIndex).Term
		// 找到冲突 Term 的开始一个 Index
		for i := prevLogIndex; i >= rf.log[0].Index; i -= 1 {
			if rf.GetLogEntry(i).Term != reply.ConflictTerm {
				break
			}
			reply.ConflictIndex = i
		}
		return nil
	}

	// 没有冲突问题，LogEntry 可以被添加
	// Leader 发送的 LogEntry 添加到当前节点
	for i := 0; i < len(leaderLogEntries); i++ {
		logIndex := prevLogIndex + i + 1
		// 冲突 LogEntry 重新赋值纠正
		if logIndex <= rf.GetLogEntry(-1).Index && rf.GetLogEntry(-1).Term != leaderLogEntries[i].Term {
			rf.log = rf.log[:rf.GetIndex(logIndex)]
		}
		// 新添加的 LogEntries 直接尾部添加
		if logIndex > rf.GetLogEntry(-1).Index {
			rf.log = append(rf.log, leaderLogEntries[i])
		}
	}

	// 更新状态信息 commitIndex
	leaderCommit := args.LeaderCommit
	if rf.commitIndex < leaderCommit {
		rf.commitIndex = Min(rf.GetLogEntry(-1).Index, leaderCommit)
		rf.applyCond.Signal()
	}

	reply.Success = true
	reply.Term = rf.currentTerm

	return nil
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	return ok
}
